package com.intct.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-11-13 16:14
 */
public class DataGenTest1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

//        tenv.executeSql("CREATE TABLE ws (" +
//                "  id INT," +
//                "  vc INT," +
//                "  pt AS PROCTIME()," +
//                "  et AS cast(CURRENT_TIMESTAMP as timestamp(3))," +
//                "  WATERMARK FOR et AS et - INTERVAL '5' SECOND" +
//                ") WITH (" +
//                "  'connector' = 'datagen'," +
//                "  'rows-per-second' = '10'," +
//                "  'fields.id.min' = '1'," +
//                "  'fields.id.max' = '3'," +
//                "  'fields.vc.min' = '1'," +
//                "  'fields.vc.max' = '100'" +
//                ")");

        String createCdcSource = "CREATE TABLE mysql_t1_cdc (" +
                "  id STRING," +
                "  name STRING," +
                "  pt AS PROCTIME()," +
                "  create_time timestamp(3)," +
                "  WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND, " +
                "  PRIMARY KEY (id) NOT ENFORCED" +
                ") WITH (" +
                "  'connector' = 'mysql-cdc'," +
                "  'hostname' = 'cdh-node'," +
                "  'port' = '13306'," +
                "  'username' = 'root'," +
                "  'password' = 'Test_090110'," +
                "  'database-name' = 'm1'," +
                "  'table-name' = 'test01'," +
                "  'scan.startup.mode' = 'initial'" +
                ")";
        tenv.executeSql(createCdcSource);
        
        tenv.executeSql("SELECT " +
                " window_start, " +
                " window_end, " +
                " id , " +
                " COUNT(name)" +
                " FROM TABLE(" +
                "  CUMULATE(TABLE mysql_t1_cdc, DESCRIPTOR(create_time), INTERVAL '2' SECONDS , INTERVAL '6' SECONDS))" +
                " GROUP BY window_start, window_end, id").print();

    }
}
